package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        tenv.executeSql(MyKafkaUtil.getTopicDb("order_pre_process_211126"));
//        id
//        order_id
//        sku_id
//        sku_name
//        img_url
//        order_price
//        sku_num
//        create_time
//        source_type
//        source_id
//        split_total_amount
//        split_activity_amount
//        split_coupon_amount
/**
 * {

  id": 7472,
 "consignee": "钟离邦",
 "consignee_tel": "13358423217",
 "total_amount": 6964.00,
 "order_status": "1002",
 "user_id": 171,
 "payment_way": null,
 "delivery_address": "第4大街第22号楼4单元463门",
 "order_comment": "描述224337",
 "out_trade_no": "749166429762541",
 "trade_body": "小米电视4A 70英寸 4K超高清 HDR 二级能效 2GB+16GB L70M5-4A 内置小爱 智能网络液晶平板教育电视等5件商品",
 "create_time": "2025-06-25 19:27:51",
 "operate_time": "2025-06-25 19:27:51",
 "expire_time": "2025-06-25 19:42:51",
 "process_status": null,
 "tracking_no": null,
 "parent_order_id": null,
 "img_url": "http://img.gmall.com/896931.jpg",
 "province_id": 3,
 "activity_reduce_amount": 0.00,
 "coupon_reduce_amount": 0.00,
 "original_total_amount": 6946.00,
 "feight_fee": 18.00,
 "feight_fee_reduce": null,
 "refundable_time": null
  */

        Table table1 = tenv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['sku_name'] sku_name ,\n" +
                "`data`['img_url'] img_url ,\n" +
                "`data`['order_price'] order_price,\n" +
                "`data`['sku_num'] sku_num,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['source_type'] source_type,\n" +
                "`data`['source_id'] source_id,\n" +
                "`data`['split_total_amount'] split_total_amount,\n" +
                "`data`['split_activity_amount'] split_activity_amount,\n" +
                "`data`['split_coupon_amount'] split_coupon_amount,\n" +
                "pt \n" +
                " from topic_db where `database`='gmall' and `table`='order_detail' and `type`='insert' ");
        tenv.createTemporaryView("order_detail_table",table1);


        Table table2 = tenv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount ,\n" +
                "`data`['order_status'] order_status ,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['delivery_address'] delivery_address,\n" +
                "`data`['order_comment'] order_comment,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['process_status'] process_status,\n" +
                "`data`['tracking_no'] tracking_no,\n" +
                "`data`['parent_order_id'] parent_order_id,\n" +
                "`data`['img_url'] img_url,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['feight_fee'] feight_fee,\n" +
                "`data`['feight_fee_reduce'] feight_fee_reduce,\n" +
                "`data`['refundable_time'] refundable_time,\n" +
                " `type`, \n" +
                " `old` \n" +
                " from topic_db where `database`='gmall' and `table`='order_info' and (`type`='insert' or `type`='update') ");
        tenv.createTemporaryView("order_info_table",table2);



        Table table3 = tenv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['order_detail_id'] order_detail_id,\n" +
                "`data`['activity_id'] activity_id ,\n" +
                "`data`['activity_rule_id'] activity_rule_id ,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['create_time'] create_time\n" +
                " from topic_db where `database`='gmall' and `table`='order_detail_activity' and `type`='insert' ");
        tenv.createTemporaryView("order_activity_table",table3);


        Table table4= tenv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['order_detail_id'] order_detail_id,\n" +
                "`data`['coupon_id'] coupon_id ,\n" +
                "`data`['coupon_use_id'] coupon_use_id ,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['create_time'] create_time\n" +
                " from topic_db where `database`='gmall'" +
                " and `table`='order_detail_coupon' and `type`='insert' ");
        tenv.createTemporaryView("order_coupon_table",table4);




        //创建base_dic LookUp表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //关联5张表
        Table table6= tenv.sqlQuery("" +
                "select " +
                "   od.id, " +
                "   od.order_id, " +
                "   od.sku_id, " +
                "   od.sku_name, " +
                "   od.order_price, " +
                "   od.sku_num, " +
                "   od.create_time, " +
                "   od.source_type source_type_id, " +
                "   dic.dic_name source_type_name, " +
                "   od.source_id, " +
                "   od.split_total_amount, " +
                "   od.split_activity_amount, " +
                "   od.split_coupon_amount, " +
                "   oi.consignee, " +
                "   oi.consignee_tel, " +
                "   oi.total_amount, " +
                "   oi.order_status, " +
                "   oi.user_id, " +
                "   oi.payment_way, " +
                "   oi.delivery_address, " +
                "   oi.order_comment, " +
                "   oi.out_trade_no, " +
                "   oi.trade_body, " +
                "   oi.operate_time, " +
                "   oi.expire_time, " +
                "   oi.process_status, " +
                "   oi.tracking_no, " +
                "   oi.parent_order_id, " +
                "   oi.province_id, " +
                "   oi.activity_reduce_amount, " +
                "   oi.coupon_reduce_amount, " +
                "   oi.original_total_amount, " +
                "   oi.feight_fee, " +
                "   oi.feight_fee_reduce, " +
                "   oi.refundable_time, " +
                "   oa.id order_detail_activity_id, " +
                "   oa.activity_id, " +
                "   oa.activity_rule_id, " +
                "   oc.id order_detail_coupon_id, " +
                "   oc.coupon_id, " +
                "   oc.coupon_use_id, " +
                "   oi.`type`, " +
                "   oi.`old`, " +
                "   od.pt," +
                "current_row_timestamp() row_op_ts " +
                "from order_detail_table od " +
                "join order_info_table oi " +
                "on od.order_id=oi.id " +
                "left join order_activity_table oa " +
                "on od.id = oa.order_detail_id " +
                "left join order_coupon_table oc " +
                "on od.id=oc.order_detail_id " +
                "join base_dic FOR SYSTEM_TIME AS OF od.pt as dic " +
                "on od.source_type=dic.dic_code");
        tenv.createTemporaryView("result_table",table6);

        tenv.toRetractStream(table6,Row.class).print(">>>>>>");
        //创建upsert-kafka表
        //使用下面这句来将user_action_time 声明为事件时间,并且声明watermark 的生成规则，即 user_action_time 减 5 秒
        //事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型
        tenv.executeSql("CREATE TABLE dwd_order_pre (\n" +
                " `id` STRING,\n" +
                "  `order_id` STRING,\n" +
                "  `sku_id` STRING,\n" +
                "  `sku_name` STRING,\n" +
                "  `order_price` STRING,\n" +
                "  `sku_num` STRING,\n" +
                "  `create_time` STRING,\n" +
                "  `source_type_id` STRING,\n" +
                "  `source_type_name` string,\n" +
                "  `source_id` string,\n" +
                "  `split_total_amount` string,\n" +
                "  `split_activity_amount` string,\n" +
                "  `split_coupon_amount` string,\n" +
                "  `consignee` string,\n" +
                "  `consignee_tel` string,\n" +
                "  `total_amount` string,\n" +
                "  `order_status` string,\n" +
                "  `user_id` string,\n" +
                "  `payment_way` string,\n" +
                "  `delivery_address` string,\n" +
                "  `order_comment` string,\n" +
                "  `out_trade_no` string,\n" +
                "  `trade_body` string,\n" +
                "  `operate_time` string,\n" +
                "  `expire_time` string,\n" +
                "  `process_status` string,\n" +
                "  `tracking_no` string,\n" +
                "  `parent_order_id` string,\n" +
                "  `province_id` string,\n" +
                "  `activity_reduce_amount` string,\n" +
                "  `coupon_reduce_amount` string,\n" +
                "  `original_total_amount` string,\n" +
                "  `feight_fee` string,\n" +
                "  `feight_fee_reduce` string,\n" +
                "  `refundable_time` string,\n" +
                "  `order_detail_activity_id` string,\n" +
                "  `activity_id` string,\n" +
                "  `activity_rule_id` string,\n" +
                " `order_detail_coupon_id` string,\n" +
                " `coupon_id` string,\n" +
                " `coupon_use_id` string,\n" +
                " `type` string,\n" +
                " `old` map<string,string>,\n" +
                " `pt` TIMESTAMP_LTZ(3),\n" +
                "  row_op_ts TIMESTAMP_LTZ(3),\n" +
                "  PRIMARY KEY (id) NOT ENFORCED \n" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));
            //将数据写出
        tenv.executeSql("insert into dwd_order_pre select * from result_table");

        //启动任务
        env.execute("DwdTradeOrderPreProcess");













        env.execute();

    }
}
